package ooo.reindeer.concurrent.netqueue.handler;

import io.netty.channel.ChannelHandlerContext;
import ooo.reindeer.concurrent.netqueue.message.Command;
import ooo.reindeer.concurrent.netqueue.message.Message;
import ooo.reindeer.logging.ILogger;
import ooo.reindeer.logging.LogUtil;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class MessagePushHandler implements Runnable {

    private static final ILogger logger = LogUtil.getLogger(MessagePushHandler.class);

    Message message;
    ChannelHandlerContext channel;
    BlockingQueue<byte[]> queue;
    boolean isExit = false;
    Map<String, MessagePushHandler> messagePushHandlerMap;

    public MessagePushHandler(Message message, ChannelHandlerContext channel, BlockingQueue<byte[]> queue, Map<String, MessagePushHandler> messagePushHandlerMap) {
        this.message = message;
        this.channel = channel;
        this.queue = queue;
        this.messagePushHandlerMap = messagePushHandlerMap;
        this.messagePushHandlerMap.put(String.format("%s-%s", channel, message.getTopic()), this);
    }

    public void exit() {
        isExit = true;
    }

    @Override
    public void run() {
        logger.info("topic [{}] push handler for [{}] enter", message.getTopic(), channel);
        while ((!channel.isRemoved()) && (!isExit)) {
            logger.debug("topic [{}] poll", message.getTopic());
            try {

                byte[] data = queue.poll(1, TimeUnit.SECONDS);
                if (data != null) {
                    Message push = new Message();
                    push.setCommand(Command.PUSH);
                    push.setTopic(message.getTopic());
                    push.setData(data);
                    logger.debug("Push [{}]", push);
                    channel.writeAndFlush(push);
                }
            } catch (Exception e) {

                logger.info("", e);
            }
        }
        messagePushHandlerMap.put(String.format("%s-%s", channel, message.getTopic()), this);
        logger.info("topic [{}] push handler for [{}] exit", message.getTopic(), channel);
    }
}
